package com.tandashi.springbootstudy01.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.reactor.IOReactorException;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.Assert;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;

@Configuration
public class ElasticConfig {


    @Bean
    @Primary
    public PoolingNHttpClientConnectionManager poolingNHttpClientConnectionManager() throws IOReactorException {
        DefaultConnectingIOReactor reactor = new DefaultConnectingIOReactor(
                IOReactorConfig.custom()
                        .setSoKeepAlive(true)
                        .setSelectInterval(30000)
                        .setSoReuseAddress(true)
                        .setConnectTimeout(this.connectTimeout)
                        .setSoTimeout(this.socketTimeout)
                        .build());

        PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(reactor);
        //每个Host最大并发连接数
        connectionManager.setDefaultMaxPerRoute(this.defaultMaxPerRoute);
        //最大连接数
        connectionManager.setMaxTotal(this.maxTotal);

        return connectionManager;
    }

    static HttpHost[] buildHosts(String hosts) {
        return Arrays.stream(StringUtils.split(hosts, ','))
                .map(ElasticConfig::buildHttpHost)
                .toArray(HttpHost[]::new);
    }

    static HttpHost buildHttpHost(String hostAndPort) {
        Assert.notNull(hostAndPort, "HostAndPort is null");
        String[] address = StringUtils.split(hostAndPort, ':');
        return new HttpHost(address[0], Integer.parseInt(address[1]), HTTP_SCHEME);
    }

    @Bean
    @Primary
    public RestHighLevelClient restHighLevelClient(PoolingNHttpClientConnectionManager connectionManager) {
        HttpHost[] hosts = buildHosts(nodes);
        RestClientBuilder clientBuilder = RestClient.builder(hosts);

        clientBuilder.setMaxRetryTimeoutMillis(this.maxRetryTimeout);
        clientBuilder.setRequestConfigCallback(
                builder -> builder
                        .setConnectTimeout(this.connectTimeout)
                        .setSocketTimeout(this.socketTimeout));

        clientBuilder.setHttpClientConfigCallback(httpAsyncClientBuilder -> {

            if (StringUtils.isNotBlank(this.username) && StringUtils.isNotBlank(this.password)) {

                final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();

                credentialsProvider.setCredentials(
                        AuthScope.ANY,
                        new UsernamePasswordCredentials(this.username, this.password));

                httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }

            return httpAsyncClientBuilder.setConnectionManager(connectionManager);
        });

        return new RestHighLevelClient(clientBuilder);
    }

    @Scheduled(cron = "0,30 * * * * ?")
    public void closeExpiredAndIdleConnections() throws IOReactorException {
        PoolingNHttpClientConnectionManager manager = poolingNHttpClientConnectionManager();
        manager.closeExpiredConnections();
        manager.closeIdleConnections(30, TimeUnit.SECONDS);
    }

    /**
     * 主机名数组，支持逗号分隔（host:port）
     */
    @Value("${elastic.nodes}")
    private String nodes;

    /**
     * 最大重试超时时长
     */
    @Value("${elastic.maxRetryTimeout}")
    private int maxRetryTimeout;

    /**
     * 连接超时时间
     */
    @Value("${elastic.connectTimeout}")
    private int connectTimeout;

    /**
     * Socket超时时长
     */
    @Value("${elastic.socketTimeout}")
    private int socketTimeout;

    /**
     * 用户名
     */
    @Value("${elastic.username:elastic}")
    private String username;

    /**
     * 密码
     */
    @Value("${elastic.password:123456}")
    private String password;

    /**
     * 每个Host最大并发连接数
     */
    @Value("${elastic.defaultMaxPerRoute}")
    private int defaultMaxPerRoute;

    /**
     * 最大连接数
     */
    @Value("${elastic.maxTotal}")
    private int maxTotal;

    static final String HTTP_SCHEME = "http";
}
